/**
 * 
 */
package config;

import mapper.SparseMatrixBuilderMapper;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import reducer.SparseMatrixBuilderReducer;

/**
 * @author puneet
 *
 */
public class SparseMatrixBuilderJobRunner {

	/**
	 * @param args
	 */
	public static void main(String[] args)throws Exception {
		Configuration conf = new Configuration();
		Path ipFilePath = new Path(args[0]);
		conf.setInt("sourceIndex", Integer.parseInt(args[2]));
		conf.setInt("destIndex", Integer.parseInt(args[3]));
		Job job = new Job(conf,"sparsematrixbuilder");
		job.setMapperClass(SparseMatrixBuilderMapper.class);
		job.setReducerClass(SparseMatrixBuilderReducer.class);
		job.setOutputKeyClass(LongWritable.class);
		job.setOutputValueClass(Text.class);
		job.setCombinerClass(SparseMatrixBuilderReducer.class);
		TextInputFormat.addInputPath(job, ipFilePath);
		FileOutputFormat.setOutputPath(job, new Path(args[1]));
		job.setJarByClass(SparseMatrixBuilderJobRunner.class);
		job.setNumReduceTasks(40);
		
		job.waitForCompletion(true);
	}

}
